/**
 * Software License, Version 1.0 Copyright 2003 The Trustees of Indiana
 * University. All rights reserved. Redistribution and use in source and binary
 * forms, with or without modification, are permitted provided that the
 * following conditions are met: 1) All redistributions of source code must
 * retain the above copyright notice, the list of authors in the original source
 * code, this list of conditions and the disclaimer listed in this license; 2)
 * All redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the disclaimer listed in this license in the
 * documentation and/or other materials provided with the distribution; 3) Any
 * documentation included with all redistributions must include the following
 * acknowledgement: "This product includes software developed by the Community
 * Grids Lab. For further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/." Alternatively, this acknowledgement may
 * appear in the software itself, and wherever such third-party acknowledgments
 * normally appear. 4) The name Indiana University or Community Grids Lab or
 * NaradaBrokering, shall not be used to endorse or promote products derived
 * from this software without prior written permission from Indiana University.
 * For written permission, please contact the Advanced Research and Technology
 * Institute ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5)
 * Products derived from this software may not be called NaradaBrokering, nor
 * may Indiana University or Community Grids Lab or NaradaBrokering appear in
 * their name, without prior written permission of ARTI. Indiana University
 * provides no reassurances that the source code provided does not infringe the
 * patent or any other intellectual property rights of any other entity. Indiana
 * University disclaims any liability to any recipient for claims brought by any
 * other entity based on infringement of intellectual property rights or
 * otherwise. LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH
 * NO WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY
 * GIVES NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.samples.secureAV;

/**
 * This class is adapted from cgl.narada.samples.rtp.NBRTPConnector. This class
 * does not use RTPEvent. Instead it uses native NBEvent to transmit and receive
 * the stream.
 * 
 * @author Hasan Bulut - hbulut@indiana.edu Community Grids Lab., Indiana
 *         University
 * @version 1.0
 */

import java.io.IOException;
import java.security.PrivateKey;
import java.security.PublicKey;
import java.util.Properties;

import javax.media.protocol.ContentDescriptor;
import javax.media.protocol.PushSourceStream;
import javax.media.protocol.SourceTransferHandler;
import javax.media.rtp.OutputDataStream;
import javax.media.rtp.RTPConnector;

import cgl.narada.event.EventProperties;
import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventException;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.TemplateInfoImpl;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.ConsumerConstraints;
import cgl.narada.service.qos.ProducerConstraints;
import cgl.narada.service.security.kmc.messages.SecureTopicKeyResponse;
import cgl.narada.transport.TransportHandler;
import cgl.narada.util.SystemInit;
import cgl.narada.util.SystemNotInitializedException;

public class NBMediaConnector implements RTPConnector, NBEventListener {
	private static ClientService clientService;
	private static boolean clientServiceInitialized = false;
	private static boolean tdsEnabled = false;

	private String ip; // broker ip
	private int brokerPort; // broker port
	private int localPort;
	private int receiveBufferSize;
	private int sendBufferSize;
	private NBInputStream dataInputStream = null;
	private NBInputStream controlInputStream = null;
	private NBOutputStream dataOutputStream = null;
	private NBOutputStream controlOutputStream = null;
	protected TransportHandler transportHandler;
	private String dataTopic = "Media/Data";
	private String controlTopic = "Media/Control";
	private int entityID;

	private EventConsumer receiver;
	private EventProducer transmitter;
	// private int templateID = 3142;
	private EventProperties evtProp;
	private Integer evtID;
	private String moduleName = "NB-NATIVE: NBMediaConnector: ";

	private boolean debug = false;
	// private boolean withTDS = false;
	private boolean publishEnabled = true;
	private int tracker = 0;

	private static int eventCounter = 0;

	public NBMediaConnector(String ip, int brkPort, int entityID, int topicNo) {
		this.ip = ip;
		this.brokerPort = brkPort;
		dataTopic = dataTopic + "/" + topicNo;
		int ctrlNo = topicNo + 1;
		controlTopic = controlTopic + "/" + ctrlNo;
		this.entityID = entityID;
		evtID = new Integer(entityID);
		try {
			String serviceConfigPath = null;
			try {
				serviceConfigPath = SystemInit
						.getInitData(SystemInit.SYS_SERVICE_CONFIG_PATH);
			}
			catch (SystemNotInitializedException e) {
				e.printStackTrace();
			}

			SessionService.setServiceConfigurationLocation(serviceConfigPath);
			clientService = SessionService.getClientService(entityID);
		}
		catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public NBMediaConnector(String ip, int brkPort, int entityID) {
		this.ip = ip;
		this.brokerPort = brkPort;
		this.entityID = entityID;
		evtID = new Integer(entityID);
		try {
			String serviceConfigPath = null;
			try {
				serviceConfigPath = SystemInit
						.getInitData(SystemInit.SYS_SERVICE_CONFIG_PATH);
			}
			catch (SystemNotInitializedException e) {
				e.printStackTrace();
			}

			SessionService.setServiceConfigurationLocation(serviceConfigPath);
			clientService = SessionService.getClientService(entityID);
		}
		catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public void initializeBrokerCommunications(Properties props, String commType)
			throws ServiceException {
		if (!clientServiceInitialized) {
			clientService.initializeBrokerCommunications(props, commType);
			clientServiceInitialized = true;
		}
		receiver = clientService.createEventConsumer(this);
	}

	/**
	 * @return Returns the controlTopic.
	 */
	public String getControlTopic() {
		return controlTopic;
	}

	/**
	 * @return Returns the dataTopic.
	 */
	public String getDataTopic() {
		return dataTopic;
	}

	// HG: Modified methods for creating consumer constraints...

	public void subscribe(String _data, SecureTopicKeyResponse r2, PublicKey root)
			throws ServiceException {

		Profile profile1 = clientService.createProfile(
				TemplateProfileAndSynopsisTypes.STRING, _data);

		ConsumerConstraints constraints = receiver
				.createConsumerConstraints(profile1);
		constraints.setReceiveSecurely(entityID);
		constraints.setSecurityToken(r2.getSignedSecurityToken(), r2.getKey(),
				null, false, root);

		if (tdsEnabled) {
			constraints.setReceiveInTimeOrder();
			constraints.setReceiveAfterTimeSpacing();
		}

		receiver.subscribeTo(profile1, constraints);

		encryptedDataTopicUUID = _data;
	}

	public void enableTDS() {
		tdsEnabled = true;
	}

	public void initializeTransmitter(String sig, String mode, String padding,
			String _data, SecureTopicKeyResponse dataToken, String _control,
			SecureTopicKeyResponse controlToken, PrivateKey priv)
			throws ServiceException {
		transmitter = clientService.createEventProducer();
		transmitter.setSuppressRedistributionToSource(true);
		transmitter.generateEventIdentifier(true);
		transmitter.setHighResolutionTimestamp(true);
		transmitter.setDisableTimestamp(false);

		// HG: Added producer constraints...

		Properties encryptProps = new Properties();
		encryptProps.put(ProducerConstraints.SIGNING_ALGORITHM,
				(sig == null ? "SHA1withRSA" : sig));
		encryptProps.put(ProducerConstraints.CIPHER_MODE, (mode == null ? "CBC"
				: mode));
		encryptProps.put(ProducerConstraints.CIPHER_PADDING,
				(padding == null ? "PKCS7Padding" : padding));

		// for data
		encryptedDataTopicUUID = _data;

		TemplateInfo tid = new TemplateInfoImpl(12345,
				TemplateProfileAndSynopsisTypes.STRING, _data);

		pcData = transmitter.createProducerConstraints(tid);
		pcData.setSendSecurely();
		pcData.setSecurityToken(dataToken.getSignedSecurityToken(), dataToken
				.getKey(), priv, encryptProps);

		// System.out.println("Created PC[" + (pcData == null) + "]" + _data);

		// for control topic
		encryptedControlTopicUUID = _control;

		TemplateInfo tic = new TemplateInfoImpl(67890,
				TemplateProfileAndSynopsisTypes.STRING, _control);

		pcControl = transmitter.createProducerConstraints(tid);
		pcControl.setSendSecurely();
		pcControl.setSecurityToken(controlToken.getSignedSecurityToken(),
				controlToken.getKey(), priv, encryptProps);

		// System.out
		// .println("Created PC[" + (pcControl == null) + "]" + _control);
	}

	/**
	 * @param encryptedControlTopicUUID
	 *            The encryptedControlTopicUUID to set.
	 */
	public void setEncryptedControlTopicUUID(String encryptedControlTopicUUID) {
		this.encryptedControlTopicUUID = encryptedControlTopicUUID;
	}

	/**
	 * @param encryptedDataTopicUUID
	 *            The encryptedDataTopicUUID to set.
	 */
	public void setEncryptedDataTopicUUID(String encryptedDataTopicUUID) {
		this.encryptedDataTopicUUID = encryptedDataTopicUUID;
	}

	ProducerConstraints pcData;
	ProducerConstraints pcControl;

	String encryptedDataTopicUUID;
	String encryptedControlTopicUUID;

	public void onEvent(NBEvent nbEvent) {

		// System.out
		// .println("Recieved event on: " + nbEvent.getContentSynopsis());

		int eventType = nbEvent.getEventType();

		if (eventType == NBEventTypes.NORMAL_EVENT) {
			String synopsis = (String) nbEvent.getContentSynopsis();
			if (debug) {
				System.out.println(moduleName + " " + eventCounter
						+ " Received NBEvent ");
				eventCounter++;
			}

			if (synopsis.equals(encryptedDataTopicUUID)) {
				if (dataInputStream != null) {
					dataInputStream.onMessage(nbEvent);
					// dataInputStream.onMessage(replayEvent.getNBEvent());
				}
				else {
					System.err.println(this.getClass().getName()
							+ ": Data received but dataInputStream is null.");
				}
				return;
			}
			else if (synopsis.equals(encryptedControlTopicUUID)) {
				if (controlInputStream != null) {
					controlInputStream.onMessage(nbEvent);
				}
				else {
					System.err
							.println(this.getClass().getName()
									+ ": Data received but controlInputStream is null.");
				}
				return;
			}
		}
	}

	/** Enable or diable publishing of streams */
	public void setPublishEnabled(boolean toPublish) {
		publishEnabled = toPublish;
	}

	public void publishEvent(byte[] data, int offset, int length, String tpcName)
			throws ServiceException, NBEventException {

		if (transmitter == null) {
			// System.out.println("null transmitter");
			// System.exit(0);
			return;
		}

		if (publishEnabled) {

			NBEvent nbEvent = transmitter.generateEvent(
					TemplateProfileAndSynopsisTypes.STRING, tpcName, data);

			if (tpcName.endsWith(encryptedDataTopicUUID))
				transmitter.publishEvent(nbEvent, pcData);
			else
				transmitter.publishEvent(nbEvent, pcControl);
			tracker++;
			// System.out.println(moduleName + "Published (" + tracker
			// + ") events on " + tpcName);
		}
		/* end if (publishEnabled) */

	}

	public boolean isClient() {
		return true;
	}

	public void close() {
		try {
			closeBrokerConnection();
		}
		catch (ServiceException ex) {
			ex.printStackTrace();
		}
	}

	public void closeBrokerConnection() throws ServiceException {
		clientService.closeBrokerConnection();
		clientService.terminateServices();
	}

	/**
	 * Returns an input stream to receive the RTP data.
	 */
	public PushSourceStream getDataInputStream() {
		if (dataInputStream == null) {
			// dataInputStream = new NBInputStream(dataTopic);
			dataInputStream = new NBInputStream(encryptedDataTopicUUID);
		}
		return dataInputStream;
	}

	/**
	 * Returns an output stream to send the RTP data.
	 */
	public OutputDataStream getDataOutputStream() {
		if (dataOutputStream == null) {
			// dataOutputStream = new NBOutputStream(dataTopic);
			dataOutputStream = new NBOutputStream(encryptedDataTopicUUID);
		}
		return dataOutputStream;
	}

	/**
	 * Returns an input stream to receive the RTCP data.
	 */
	public PushSourceStream getControlInputStream() {
		if (controlInputStream == null) {
			// controlInputStream = new NBInputStream(controlTopic);
			controlInputStream = new NBInputStream(encryptedControlTopicUUID);
		}
		return controlInputStream;
	}

	/**
	 * Returns an output stream to send the RTCP data.
	 */
	public OutputDataStream getControlOutputStream() {
		if (controlOutputStream == null) {
			// controlOutputStream = new NBOutputStream(controlTopic);
			controlOutputStream = new NBOutputStream(encryptedControlTopicUUID);
		}

		return controlOutputStream;
	}

	/**
	 * Set the receive buffer size of the RTP data channel. This is only a hint
	 * to the implementation. The actual implementation may not be able to do
	 * anything to this.
	 */
	public void setReceiveBufferSize(int size) throws IOException {
		receiveBufferSize = size;
	}

	/**
	 * Get the receive buffer size set on the RTP data channel. Return -1 if the
	 * receive buffer size is not applicable for the implementation.
	 */
	public int getReceiveBufferSize() {
		return receiveBufferSize;
	}

	/**
	 * Set the send buffer size of the RTP data channel. This is only a hint to
	 * the implementation. The actual implementation may not be able to do
	 * anything to this.
	 */
	public void setSendBufferSize(int size) throws IOException {
		sendBufferSize = size;
	}

	/**
	 * Get the send buffer size set on the RTP data channel. Return -1 if the
	 * send buffer size is not applicable for the implementation.
	 */
	public int getSendBufferSize() {
		return sendBufferSize;
	}

	/**
	 * Return the RTCP bandwidth fraction. This value is used to initialize the
	 * RTPManager. Check RTPManager for more detauls. Return -1 to use the
	 * default values.
	 */
	public double getRTCPBandwidthFraction() {
		return -1;
	}

	/**
	 * Return the RTCP sender bandwidth fraction. This value is used to
	 * initialize the RTPManager. Check RTPManager for more detauls. Return -1
	 * to use the default values.
	 */
	public double getRTCPSenderBandwidthFraction() {
		return -1;
	}

	class NBOutputStream implements OutputDataStream {
		private String topicName;
		boolean isDataStream;

		public NBOutputStream(String tpcName) {
			this.topicName = tpcName;
			System.out.println("TOPIC NAME: " + tpcName);
			if (topicName.equalsIgnoreCase(encryptedDataTopicUUID)) {
				isDataStream = true;
			}
			else {
				isDataStream = false;
			}
		}

		public int write(byte[] data, int offset, int len) {
			// send the package
			try {
				publishEvent(data, offset, len, topicName);
			}
			catch (Exception e) {
				e.printStackTrace();
				return -1;
			}
			return len;
		}
	} // end of class NBOutputStream

	class NBInputStream extends Thread implements PushSourceStream {
		SourceTransferHandler sth = null;
		NBEvent message;
		private String topicName;
		Object synchObject = new Object();
		EventProperties evtProp;
		int off = 0;
		int len = 0;

		public NBInputStream(String tpcName) {
			this.topicName = tpcName;
		}

		public int read(byte[] buffer, int offset, int length) {
			synchronized (synchObject) {
				off = 0;
				len = message.getContentPayloadSize();
				System.arraycopy(message.getContentPayload(), off, buffer,
						offset, len);
			}
			return message.getContentPayloadSize();
		}

		public void onMessage(NBEvent message) {
			synchronized (synchObject) {
				this.message = message;
			}
			if (sth != null) {
				sth.transferData(this);
			}
			else {
				System.out.println(this.getClass().getName()
						+ ": sth is null .......................");
			}
		}

		public int getMinimumTransferSize() {
			return 2 * 1024; // twice the MTU size, just to be safe.
		}

		public synchronized void setTransferHandler(SourceTransferHandler sth) {
			this.sth = sth;
		}

		// Not applicable.
		public ContentDescriptor getContentDescriptor() {
			return null;
		}

		// Not applicable.
		public long getContentLength() {
			return LENGTH_UNKNOWN;
		}

		// Not applicable.
		public boolean endOfStream() {
			return false;
		}

		// Not applicable.
		public Object[] getControls() {
			return new Object[0];
		}

		// Not applicable.
		public Object getControl(String type) {
			return null;
		}
	} // end of class NBInputStream

}
